package cn.jx.asc.mqtt;


import cn.jx.asc.mqtt.handle.MessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

import java.text.SimpleDateFormat;
import java.util.logging.Logger;

/**
 * @author z423
 * @2021-06-13-16:50
 * @effect   消息的接收
 * @especial
 */
@Slf4j
@SuppressWarnings("all")
public class PushCallback implements MqttCallback {

    private static final Logger logger = Logger.getLogger(PushCallback.class.getCanonicalName());
    private final MqClient mqClient;
    private int reConnTimes = 0;
    private final MessageHandler handler;

    public PushCallback(MqClient client, MessageHandler handler) {
        this.mqClient = client;
        this.handler = handler;
    }

    /**
     * 掉线时调用函数
     * @param cause Throwable
     */
    @Override
    public void connectionLost(Throwable cause) {

        log.info("connect is losted,and try to reconnect,cause = " + cause.getMessage());
        cause.printStackTrace();
        while (!mqClient.reConnect()) {
            try {
                if (reConnTimes++ > 20) {//前20次每秒重连一次
                    Thread.sleep(1000);
                } else {//超过20次后没10s重连一次
                    Thread.sleep(10000);
                }
            } catch (InterruptedException e) {
                log.error(e.getMessage());
            }
        }

    }

    /**
     * paho 本身有个bug，即此函数接口抛出异常都会回调到connectionLost()的接口，故需要在此函数中用try catch包起来处理，
     *   确保无异常抛出。
     * @param topic topic
     * @param message message
     */
    @Override
    public void messageArrived(String topic, MqttMessage message)       {

        try {
           // 数据 protobuf 解码
           byte[] payload = message.getPayload();
           OnenetMq.Msg obj = OnenetMq.Msg.parseFrom(payload);
            handler.handle(obj.getMsgid(), new String(obj.getData().toByteArray()));
       }catch(Exception e){
           log.info("messageArrived phrase exception: {}", e.getMessage());
       }finally {
           if(mqClient.getManualAcks()){
               mqClient.messageArrivedComplete(message);
           }
       }

    }


    /**
     * @param token IMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {

        IMqttAsyncClient client = token.getClient();
        log.info("deliveryComplete: {}", client);

    }

}
